package com.lbx;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.stereotype.Component;
import org.springframework.util.StopWatch;

/**
 * @description:
 * @author: lbx
 * @create: 2018/11/7 10:13
 **/
@Component
public class RocketMQClient {
    /**
     * 生产者的组名
     */
    private String producerGroup="Producer";

    /**
     * NameServer 地址
     */
    private String namesrvAddr ="192.168.107.101:9876";

    public void defaultMQProducer() {
        //生产者的组名
        DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
        //指定NameServer地址，多个地址以 ; 隔开
        producer.setNamesrvAddr(namesrvAddr);
        try {
            /**
             * Producer对象在使用之前必须要调用start初始化，初始化一次即可
             * 注意：切记不可以在每次发送消息时，都调用start方法
             */

            producer.start();


            //创建一个消息实例，包含 topic、tag 和 消息体
            //如下：topic 为 "TopicTest"，tag 为 "push"

            StopWatch stop = new StopWatch();
            stop.start();

            for (int i = 0; i < 100; i++) {
                Message message = new Message("TopicTest", "clickNum", (i+"").getBytes(RemotingHelper.DEFAULT_CHARSET));
                message.setKeys("omsTopic"+i);
                SendResult result = producer.send(message);
                System.out.println("发送响应：MsgId:" + result.getMsgId() + "，发送状态:" + result.getSendStatus());
            }
            stop.stop();
            System.out.println("----------------发送一万条消息耗时：" + stop.getTotalTimeMillis());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.shutdown();
        }
    }

    public static void main(String[] args) {
        new RocketMQClient().defaultMQProducer();
    }
}
